package com.example.message.service;

import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import jakarta.annotation.Resource;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

import java.util.stream.Collectors;
import java.util.stream.Stream;

@Service
public class SenderService {

    @Resource
    private StreamBridge streamBridge;

    /**
     * 发送普通消息
     * @param bindingName 主题
     * @param msg 消息
     * @return boolean
     */
    public boolean send(String bindingName, String msg) {
        return streamBridge.send(bindingName, MessageBuilder.withPayload(msg).build());
    }

    // 发送广播消息 (TAGS) : 同一消费组下所有消费者
    public <T> boolean send(String bindingName, String tag, String key, T msg) {
        return streamBridge.send(bindingName, MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_TAGS, tag) // 消息Tag，用于消费端根据指定Tag过滤消息
                .setHeader(MessageConst.PROPERTY_KEYS, key) // 消息索引键，可根据关键字精确查找某条消息
                //.setHeader(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, 1) //消息id
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
    }

    public <T> boolean sendWithTags(String bindingName, T msg, String tag) {
        Message<T> message = MessageBuilder.createMessage(msg,
                new MessageHeaders(Stream.of(tag).collect(Collectors
                        .toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
        return streamBridge.send(bindingName, message);
    }

    public <T> boolean sendObject(String bindingName, T msg, String tag) {
        Message<T> message = MessageBuilder.withPayload(msg)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        return streamBridge.send(bindingName, message);
    }

    public <T> boolean sendTransactionalMsg(String bindingName, T msg, int num) {
        Message<T> message = MessageBuilder.withPayload(msg)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .setHeader("test", String.valueOf(num))
                .setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, "binder").build();
        return streamBridge.send(bindingName, message);
    }

    /**
     * 延迟消息
     * @param bindingName 主题
     * @param msg 延时消息体
     * @param delayLevel 延时级别 1~18 (1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h)
     * @return boolean
     */
    public <T> boolean sendDelayMsg(String bindingName, T msg, int delayLevel) {
        return streamBridge.send(bindingName, MessageBuilder.withPayload(msg)
                //.setHeader("x-delay", 5 * 1000) //rabbit
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayLevel)
                .build());
    }
}
